package com.wva.video.consumer.sync2es;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.StrUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.wva.common.client.UserClient;
import com.wva.common.domain.vo.UserProfileVO;
import com.wva.common.utils.CustomObjectMapper;
import com.wva.common.utils.JsonUtils;
import com.wva.common.utils.R;
import com.wva.es.utils.DataElasticsearchUtils;
import com.wva.video.domain.CanalMessageEntity;
import com.wva.video.domain.po.Video;
import com.wva.video.elasticsearch.document.VideoDocument;
import com.wva.video.elasticsearch.repository.VideoRepository;
import com.wva.video.service.VideoMetadataService;
import com.wva.video.service.VideoTagService;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * 视频稿件表变化时会收集到这个队列，此消费者消费消息
 * @Author: hong
 * @Date: 2025/2/26 19:29:31
 */
@Component
@RabbitListener(queues = "video.canal.wva_video.video")
@Slf4j
public class VideoChangeConsumer {
    @Autowired
    VideoRepository videoRepository;
    @Autowired
    VideoTagService videoTagService;
    @Autowired
    UserClient userClient;
    @Autowired
    DataElasticsearchUtils dataElasticsearchUtils;
    @Autowired
    VideoMetadataService videoMetadataService;

    JsonUtils jsonUtils;

    @RabbitHandler
    void handle(Message message, @Payload Object payload) throws JsonProcessingException {
        try {
            if (jsonUtils == null) {
                //canal发送过来的json时间格式不是标准的ISO8601
                jsonUtils = new JsonUtils();
                jsonUtils.setObjectMapper(new CustomObjectMapper("yyyy-MM-dd HH:mm:ss"));
            }

            log.debug("视频稿件数据变化");
            String messageString = new String(message.getBody(), StandardCharsets.UTF_8);
            CanalMessageEntity canalMessageEntity = jsonUtils.parse(messageString, CanalMessageEntity.class);

            List<Map<String, String>> data = canalMessageEntity.getData();
            List<Map<String, String>> old = canalMessageEntity.getOld();

            for (int i = 0; i < data.size(); i++) {
                //根据不同的数据库记录操作类型来选择不同的方式更新到ES
                switch (canalMessageEntity.getType()) {
                    case "INSERT":
                        log.debug("数据库插入视频");
                        insertVideoToES(data.get(i));
                        break;
                    case "DELETE":
                        log.debug("数据库删除视频");
                        deleteVideoInES(data.get(i));
                        break;
                    case "UPDATE":
                        log.debug("数据库更新视频");
                        updateVideoInES(data.get(i), old.get(i));
                        break;
                    default:
                        //没有应对的操作类型
                        break;
                }
            }
        } catch (JsonProcessingException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }


    private void deleteVideoInES(Map<String, String> data) throws JsonProcessingException {
        Video newVideo = jsonUtils.convertMap(data, Video.class);
        videoRepository.deleteById(newVideo.getId());
        log.debug("删除es视频成功");
    }

    @SneakyThrows
    private void updateVideoInES(Map<String, String> data, Map<String, String> old) {
        //增量修改部分
        Map<String, Object> params = new HashMap<>();

        //拷贝新属性值到增量修改Map
        Set<String> oldKeys = old.keySet();
        for (String oldKey : oldKeys) {
            params.put(StrUtil.toCamelCase(oldKey, '_'), data.get(oldKey));
        }

        //需要用到视频id来定位文档id
        Video newVideo = jsonUtils.convertMap(data, Video.class);

        //如果更新了 表示标签更新了
        if (params.containsKey("tags_update_time")) {
            params.put("tags", listTagNamesAsArray(newVideo.getId()));
        }

        //增量修改
        dataElasticsearchUtils.update(params, String.valueOf(newVideo.getId()), "video");
        log.debug("更新es视频成功");
    }

    private void insertVideoToES(Map<String, String> data) throws JsonProcessingException {
        VideoDocument videoDocument = new VideoDocument();
        Video video = jsonUtils.convertMap(data, Video.class);
        BeanUtil.copyProperties(video, videoDocument);
        //补充视频标签
        videoDocument.setTags(listTagNamesAsArray(video.getId()));
        //补充用户昵称
        R<UserProfileVO> baseInfo = userClient.getBaseInfo(video.getUserId());
        String nickname = baseInfo.getData().getNickname();
        videoDocument.setUserNickname(nickname);


        //调用repository更新到es
        videoRepository.save(videoDocument);
        log.debug("插入es视频成功");
    }

    private String[] listTagNamesAsArray(Long videoId) {
        return videoTagService.listNameByVideoId(videoId).toArray(new String[0]);
    }
}
